Domain-Driven Design

# Most engineers model a domain like this:
class Order:
    def __init__(self, id, user_id, items, status, total):
        self.id = id
        self.user_id = user_id
        self.items = items
        self.status = status
        self.total = total

    def add_item(self, item):
        self.items.append(item)   # No validation. Any state is valid.
        self.total += item.price  # Mutation without an invariant check.

    def confirm(self):
        self.status = "confirmed"  # Called from anywhere. No guard.

Now watch what the database layer does six months later:

def ship_order(order_id):
    order = db.query("SELECT * FROM orders WHERE id = ?", order_id)
    order.status = "shipped"          # Direct mutation from outside
    order.items.append(free_gift())   # Adding an item to a shipped order
    db.save(order)                    # Invariants? There are none.

A shipped order with a new item added from a service 300 lines away from any validation. This is the system that Domain-Driven Design prevents. The domain model should be the source of truth for what is and is not allowed - not a dumb data container that any layer can corrupt.

What You Will Learn

How to build a Ubiquitous Language so engineers and domain experts mean the same things
Why @dataclass(eq=False) matters for Entities and @dataclass(frozen=True) matters for Value Objects
How to design an Aggregate that enforces its own invariants and prevents illegal state
How to keep your domain layer free of SQLAlchemy, FastAPI, and any other framework
How to use Domain Events to decouple side effects from business logic
How Bounded Contexts prevent one team's model from polluting another's
How Application Services orchestrate domain behaviour without containing business rules

Prerequisites

Comfortable with Python dataclasses, __eq__, and __hash__
Familiar with ABCs and typing.Protocol
Understand the Dependency Inversion Principle (Lesson 04)
Have designed at least one data model with SQLAlchemy or Django ORM

We use a single consistent domain throughout: an e-commerce order management system. The domain includes customers placing orders, orders containing lines for specific products, payments, and fulfilment.

Part 1 - Ubiquitous Language

Use the model as the backbone of a language. Commit the team to exercising that language relentlessly in all communication.

Eric Evans

The Communication Breakdown

Here is a conversation that destroys codebases:

Product Manager: "When a customer submits a purchase request, we need to confirm the booking." Engineer: "Right, so when the order is placed, I set the status to approved?" PM: "No - confirmed. And it is not an order yet, it is still a request." Engineer: "Got it. So PurchaseRequest.approved = True."

Three months later the code contains Order, PurchaseRequest, Booking, Cart, and Basket - all referring to overlapping concepts. No one can read the code without a mental translation dictionary.

Building the Glossary

DDD demands that the domain model and the domain experts share exactly the same words. This is called the Ubiquitous Language because it is used everywhere: in meetings, in tickets, in code, in tests.

# glossary.py
# This file IS the glossary. Every term defined here appears exactly this way
# everywhere in the codebase.
#
# Cart
#   An unconfirmed collection of items being assembled by a Customer.
#   A Cart becomes an Order when it is placed.
#
# Order
#   A confirmed intent to purchase. Created when the Customer places a Cart.
#   An Order has identity (OrderId) and a lifecycle: placed → confirmed
#   → shipped → delivered. It may be cancelled before it is shipped.
#
# OrderLine
#   An immutable snapshot of one product at a specific quantity and price
#   at the moment the Order was placed. Changing the product catalogue does
#   not change an existing OrderLine.
#
# Customer
#   A registered account that can place Orders.
#
# Placing an Order
#   The act of converting a Cart into an Order. Emits an OrderPlaced event.
#
# Confirming an Order
#   A warehouse action acknowledging fulfilment capacity. Emits OrderConfirmed.
#
# Shipping an Order
#   A logistics action recording that a courier collected the goods. Emits
#   OrderShipped with a tracking number.
#
# Cancelling an Order
#   Valid before the Order is Shipped. Emits OrderCancelled.

This file sounds trivial. In a large organisation it is the most important artefact in the repository. Every time an engineer and a PM disagree on a term you resolve it here and update the code to match.

What Bad Naming Costs

# Before - six different words for the same concept, one per team:
class PurchaseRequest:  pass   # PM's word
class ShoppingBasket:   pass   # UI team's word
class CartEntity:       pass   # Backend team's word
class OrderRecord:      pass   # DBA's word
class BookingObject:    pass   # Finance team's word
class TransactionItem:  pass   # Analytics team's word

Every handoff requires mental translation. Bugs live in the translation gaps.

# After - one word per concept:
class Order: pass   # confirmed intent to purchase
class Cart:  pass   # unconfirmed item collection

Part 2 - Entities

An object defined primarily by its identity is called an Entity.

An Entity has an identity that persists through state changes. Order #4521 is the same order whether its status is placed, confirmed, or shipped. Two orders with identical items and customers are still different orders if their IDs differ.

Python Implementation

# domain/entities.py
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
import uuid


@dataclass(frozen=True)
class OrderId:
    """
    Typed identity wrapper.
    Prevents accidentally comparing order IDs with customer IDs or product IDs.
    frozen=True: immutable, hashable, safe to use as dict key.
    """
    value: str

    def __post_init__(self) -> None:
        if not self.value.strip():
            raise ValueError("OrderId cannot be empty")

    def __str__(self) -> str:
        return self.value

    @classmethod
    def generate(cls) -> "OrderId":
        return cls(value=str(uuid.uuid4()))


@dataclass(eq=False)   # ← disable dataclass __eq__; we define identity manually
class Order:
    """
    Order Entity.

    Identity: order_id - two Orders are the same iff their order_id matches,
    regardless of what other fields contain.

    State: status, lines
    Behaviour: add_line, place, confirm, ship, cancel - all enforce invariants
    and record domain events.
    """
    order_id: OrderId
    customer_id: str
    status: str = field(default="pending", init=False)
    _lines: list = field(default_factory=list, repr=False, init=False)
    _events: list = field(default_factory=list, repr=False, init=False)
    _version: int = field(default=0, repr=False, init=False)  # optimistic locking

    # ── Identity-based equality ───────────────────────────────────────────────

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Order):
            return NotImplemented
        return self.order_id == other.order_id

    def __hash__(self) -> int:
        return hash(self.order_id)

    def __repr__(self) -> str:
        return f"Order(id={self.order_id}, status={self.status!r}, lines={len(self._lines)})"

    # ── Read access (immutable views) ─────────────────────────────────────────

    @property
    def lines(self) -> tuple:
        return tuple(self._lines)   # prevents external list mutation

    @property
    def events(self) -> tuple:
        return tuple(self._events)

    def pop_events(self) -> list:
        """Collect pending domain events and clear them."""
        events = list(self._events)
        self._events.clear()
        return events


# ── Demonstration of identity semantics ──────────────────────────────────────

id_a = OrderId("order-abc")
id_b = OrderId("order-abc")
id_c = OrderId("order-xyz")

o1 = Order(order_id=id_a, customer_id="cust-1")
o2 = Order(order_id=id_b, customer_id="cust-1")  # different object, same ID
o3 = Order(order_id=id_c, customer_id="cust-1")  # different ID

assert o1 == o2       # same identity
assert o1 is not o2   # different objects in memory
assert o1 != o3       # different identity
assert {o1, o2} == {o1}  # set deduplication by identity
print("Entity identity semantics: PASSED")

Why `eq=False` Is the Key Decision

# What happens WITHOUT eq=False:
from dataclasses import dataclass

@dataclass   # auto-generates __eq__ from ALL fields
class BadOrder:
    order_id: str
    status: str

o1 = BadOrder(order_id="id-1", status="pending")
o2 = BadOrder(order_id="id-2", status="pending")   # completely different order
print(o1 == o2)   # True  ← catastrophically wrong

With the default dataclass __eq__, two distinct orders that happen to have the same status would compare as equal. An entity's identity is only its ID - never its data.

Part 3 - Value Objects

An object that represents a descriptive aspect of the domain with no conceptual identity is called a Value Object.

Money($50, USD) is interchangeable with any other Money($50, USD). There is no "which fifty dollars" - they are equal by value. Value Objects are immutable and use value-based equality.

Python Implementation

# domain/value_objects.py
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
import re


# ── Money ─────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)   # immutable + auto-generates __hash__ from all fields
class Money:
    amount: Decimal
    currency: str  # ISO 4217: "USD", "EUR", "GBP"

    def __post_init__(self) -> None:
        if not isinstance(self.amount, Decimal):
            raise TypeError(f"amount must be Decimal, got {type(self.amount).__name__}")
        if self.amount < Decimal("0"):
            raise ValueError(f"Money amount cannot be negative: {self.amount}")
        if len(self.currency) != 3 or not self.currency.isalpha():
            raise ValueError(f"Invalid ISO 4217 currency code: {self.currency!r}")
        # Normalise to 2 decimal places - use object.__setattr__ because frozen
        object.__setattr__(
            self, "amount", self.amount.quantize(Decimal("0.01"), ROUND_HALF_UP)
        )
        object.__setattr__(self, "currency", self.currency.upper())

    def __add__(self, other: "Money") -> "Money":
        self._assert_same_currency(other)
        return Money(amount=self.amount + other.amount, currency=self.currency)

    def __sub__(self, other: "Money") -> "Money":
        self._assert_same_currency(other)
        result = self.amount - other.amount
        if result < Decimal("0"):
            raise ValueError(f"Subtraction result is negative: {result}")
        return Money(amount=result, currency=self.currency)

    def __mul__(self, factor: int | Decimal) -> "Money":
        return Money(
            amount=self.amount * Decimal(str(factor)), currency=self.currency
        )

    def __str__(self) -> str:
        return f"{self.currency} {self.amount:.2f}"

    def _assert_same_currency(self, other: "Money") -> None:
        if self.currency != other.currency:
            raise ValueError(
                f"Currency mismatch: cannot operate on {self.currency} and {other.currency}. "
                "Convert to a common currency first."
            )

    @classmethod
    def zero(cls, currency: str) -> "Money":
        return cls(Decimal("0"), currency)


# ── Address ───────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Address:
    street: str
    city: str
    postal_code: str
    country_code: str   # ISO 3166-1 alpha-2: "US", "GB", "DE"

    def __post_init__(self) -> None:
        if not self.street.strip():
            raise ValueError("Street cannot be empty")
        if not self.city.strip():
            raise ValueError("City cannot be empty")
        if not self.postal_code.strip():
            raise ValueError("Postal code cannot be empty")
        if len(self.country_code) != 2 or not self.country_code.isalpha():
            raise ValueError(f"Invalid ISO 3166-1 alpha-2 country code: {self.country_code!r}")
        object.__setattr__(self, "country_code", self.country_code.upper())


# ── EmailAddress ──────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class EmailAddress:
    value: str
    _PATTERN = re.compile(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")

    def __post_init__(self) -> None:
        normalised = self.value.strip().lower()
        if not self._PATTERN.match(normalised):
            raise ValueError(f"Invalid email address: {self.value!r}")
        object.__setattr__(self, "value", normalised)

    def __str__(self) -> str:
        return self.value


# ── ProductId ─────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class ProductId:
    value: str

    def __post_init__(self) -> None:
        if not self.value.strip():
            raise ValueError("ProductId cannot be empty")


# ── OrderLine (Value Object - immutable snapshot) ─────────────────────────────

@dataclass(frozen=True)
class OrderLine:
    """
    Immutable snapshot of what was ordered at what price.
    Changing the product catalogue never changes a placed OrderLine.
    """
    product_id: ProductId
    product_name: str
    quantity: int
    unit_price: Money

    def __post_init__(self) -> None:
        if self.quantity < 1:
            raise ValueError(f"Quantity must be at least 1, got {self.quantity}")
        if not self.product_name.strip():
            raise ValueError("Product name cannot be empty")

    @property
    def line_total(self) -> Money:
        return self.unit_price * self.quantity


# ── Value Object demonstrations ───────────────────────────────────────────────

m1 = Money(Decimal("50.00"), "USD")
m2 = Money(Decimal("50.00"), "usd")   # currency normalised to "USD"
m3 = Money(Decimal("75.00"), "USD")

assert m1 == m2         # value-based equality
assert m1 is not m2     # different objects
assert m1 != m3

# Immutability
try:
    m1.amount = Decimal("99.00")  # type: ignore
except Exception as e:
    print(f"Immutability enforced: {type(e).__name__}")

# Arithmetic produces new Value Objects
total = m1 + m3
assert total == Money(Decimal("125.00"), "USD")
assert total is not m1   # original unchanged

# Currency guard
try:
    _ = m1 + Money(Decimal("50.00"), "EUR")
except ValueError as e:
    print(f"Currency guard: {e}")

# OrderLine total
line = OrderLine(
    product_id=ProductId("PROD-001"),
    product_name="Python Engineering Course",
    quantity=2,
    unit_price=Money(Decimal("49.00"), "USD"),
)
assert str(line.line_total) == "USD 98.00"
print("Value Object semantics: PASSED")

Value Object vs Entity Decision Guide

Question	Entity	Value Object
Does identity matter independent of data?	Yes	No
Can two instances with identical data be different things?	Yes	No
Is it mutable over its lifetime?	Usually	Never
Does it have a tracked lifecycle?	Yes	No
Python construct	`@dataclass(eq=False)`	`@dataclass(frozen=True)`
Example	`Order`, `Customer`, `Product`	`Money`, `Address`, `OrderLine`

Part 4 - Aggregates

An Aggregate is a cluster of associated objects treated as a unit for data changes. The Aggregate Root is the only member that outside objects may hold a reference to.

The Aggregate Root is the single entry point for all mutations. All business invariants are enforced inside the aggregate - not in service classes scattered across the codebase.

The `Order` Aggregate

# domain/order_aggregate.py
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
from domain.value_objects import Money, OrderLine, ProductId, Address
from domain.entities import OrderId


@dataclass(eq=False)
class Order:
    """
    Order Aggregate Root.

    Boundary: this Order object and all its OrderLines.

    Invariants enforced here (never in service classes):
      - Cannot add lines to a non-pending order
      - Cannot place an order with zero lines
      - Cannot cancel a shipped order
      - Total is always consistent with lines (computed, not stored separately)
      - Status transitions follow the allowed lifecycle
    """
    order_id: OrderId
    customer_id: str
    shipping_address: Address
    currency: str
    status: str = field(default="pending", init=False)
    _lines: list[OrderLine] = field(default_factory=list, repr=False, init=False)
    _events: list = field(default_factory=list, repr=False, init=False)
    _version: int = field(default=0, repr=False, init=False)

    _VALID_TRANSITIONS: dict[str, set[str]] = field(
        default_factory=lambda: {
            "pending":   {"placed", "cancelled"},
            "placed":    {"confirmed", "cancelled"},
            "confirmed": {"shipped", "cancelled"},
            "shipped":   set(),          # terminal - no further transitions
            "cancelled": set(),          # terminal
        },
        repr=False,
        init=False,
        compare=False,
    )

    # ── Identity ──────────────────────────────────────────────────────────────

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Order):
            return NotImplemented
        return self.order_id == other.order_id

    def __hash__(self) -> int:
        return hash(self.order_id)

    # ── Read side ─────────────────────────────────────────────────────────────

    @property
    def lines(self) -> tuple[OrderLine, ...]:
        return tuple(self._lines)

    @property
    def total(self) -> Money:
        if not self._lines:
            return Money.zero(self.currency)
        result = Money.zero(self.currency)
        for line in self._lines:
            result = result + line.line_total
        return result

    @property
    def line_count(self) -> int:
        return len(self._lines)

    @property
    def version(self) -> int:
        return self._version

    def pop_events(self) -> list:
        events = list(self._events)
        self._events.clear()
        return events

    # ── Internal helpers ──────────────────────────────────────────────────────

    def _transition_to(self, new_status: str) -> None:
        allowed = self._VALID_TRANSITIONS.get(self.status, set())
        if new_status not in allowed:
            raise ValueError(
                f"Invalid transition: {self.status!r} → {new_status!r}. "
                f"Allowed from {self.status!r}: {allowed or 'none (terminal state)'}."
            )
        self.status = new_status
        self._version += 1

    def _require_status(self, *statuses: str) -> None:
        if self.status not in statuses:
            raise ValueError(
                f"Operation requires order to be in {statuses}; "
                f"current status is {self.status!r}."
            )

    # ── Mutating behaviour ────────────────────────────────────────────────────

    def add_line(
        self,
        product_id: str,
        product_name: str,
        quantity: int,
        unit_price: Decimal,
    ) -> None:
        self._require_status("pending")
        line = OrderLine(
            product_id=ProductId(product_id),
            product_name=product_name,
            quantity=quantity,
            unit_price=Money(unit_price, self.currency),
        )
        # Replace existing line for same product, or append
        for i, existing in enumerate(self._lines):
            if existing.product_id.value == product_id:
                # Immutable - create new line with combined quantity
                combined = OrderLine(
                    product_id=existing.product_id,
                    product_name=existing.product_name,
                    quantity=existing.quantity + quantity,
                    unit_price=existing.unit_price,
                )
                self._lines[i] = combined
                self._version += 1
                return
        self._lines.append(line)
        self._version += 1

    def remove_line(self, product_id: str) -> None:
        self._require_status("pending")
        before = len(self._lines)
        self._lines = [l for l in self._lines if l.product_id.value != product_id]
        if len(self._lines) == before:
            raise ValueError(f"Product {product_id!r} not found in order.")
        self._version += 1

    def place(self) -> None:
        if not self._lines:
            raise ValueError(
                "Cannot place an order with no items. Add at least one product first."
            )
        self._transition_to("placed")
        from domain.events import OrderPlaced
        self._events.append(
            OrderPlaced(
                order_id=str(self.order_id),
                customer_id=self.customer_id,
                total_amount=str(self.total.amount),
                currency=self.currency,
            )
        )

    def confirm(self) -> None:
        self._transition_to("confirmed")
        from domain.events import OrderConfirmed
        self._events.append(OrderConfirmed(order_id=str(self.order_id)))

    def ship(self, tracking_number: str) -> None:
        if not tracking_number.strip():
            raise ValueError("Tracking number is required to ship an order.")
        self._transition_to("shipped")
        from domain.events import OrderShipped
        self._events.append(
            OrderShipped(
                order_id=str(self.order_id),
                tracking_number=tracking_number,
            )
        )

    def cancel(self, reason: str) -> None:
        if not reason.strip():
            raise ValueError("A reason is required to cancel an order.")
        previous = self.status
        self._transition_to("cancelled")
        from domain.events import OrderCancelled
        self._events.append(
            OrderCancelled(
                order_id=str(self.order_id),
                reason=reason,
                previous_status=previous,
            )
        )

Invariant Enforcement Demo

from domain.order_aggregate import Order
from domain.entities import OrderId
from domain.value_objects import Address
from decimal import Decimal

addr = Address("1 Main St", "London", "EC1A 1AA", "GB")
order = Order(order_id=OrderId.generate(), customer_id="cust-1",
              shipping_address=addr, currency="GBP")

# Add some lines
order.add_line("P1", "Python Course", 1, Decimal("49.00"))
order.add_line("P2", "AI Systems Course", 1, Decimal("99.00"))
print(f"Total: {order.total}")   # GBP 148.00

# Place the order
order.place()
print(f"Status: {order.status}")  # placed

# Cannot add items after placing
try:
    order.add_line("P3", "New Course", 1, Decimal("29.00"))
except ValueError as e:
    print(f"Invariant enforced: {e}")

# Cannot cancel a shipped order
order.confirm()
order.ship("TRACK-UK-12345")
try:
    order.cancel("Changed mind")
except ValueError as e:
    print(f"Invariant enforced: {e}")

The critical insight: you can never reach an invalid state by calling the aggregate's public methods. Invalid states are enforced in one place - not sprinkled across service classes.

Part 5 - Repositories

A Repository mediates between the domain and the data mapping layers using a collection-like interface for accessing domain objects.

The key rule: no framework imports in the domain layer. The domain defines an abstract interface. The infrastructure implements it. The domain never knows whether storage is PostgreSQL, SQLite, or in-memory.

Abstract Repository (Domain Layer)

# domain/repositories.py
# This file has zero infrastructure imports.
from __future__ import annotations
from typing import Optional, Protocol
from domain.order_aggregate import Order


class OrderRepository(Protocol):
    """
    Abstract collection of Order aggregates.
    Implementations live in the infrastructure layer.
    """
    def get(self, order_id: str) -> Optional[Order]: ...
    def save(self, order: Order) -> None: ...
    def next_id(self) -> str: ...
    def find_by_customer(self, customer_id: str) -> list[Order]: ...

Infrastructure - SQLAlchemy Repository

# infrastructure/sqlalchemy_order_repo.py
# This file lives in infrastructure. Imports SQLAlchemy freely.
# The domain layer never imports this module.
from __future__ import annotations
from typing import Optional
from decimal import Decimal
import uuid
import json

from sqlalchemy import Column, String, Integer, Text, create_engine, select
from sqlalchemy.orm import declarative_base, Session

from domain.order_aggregate import Order
from domain.entities import OrderId
from domain.value_objects import Address, Money, OrderLine, ProductId

Base = declarative_base()


class OrderORM(Base):
    __tablename__ = "orders"
    order_id          = Column(String,   primary_key=True)
    customer_id       = Column(String,   nullable=False)
    status            = Column(String,   nullable=False, default="pending")
    currency          = Column(String(3), nullable=False)
    shipping_street   = Column(String,   nullable=False)
    shipping_city     = Column(String,   nullable=False)
    shipping_postal   = Column(String,   nullable=False)
    shipping_country  = Column(String(2), nullable=False)
    lines_json        = Column(Text,     nullable=False, default="[]")
    version           = Column(Integer,  nullable=False, default=0)


def _to_domain(row: OrderORM) -> Order:
    address = Address(
        street=row.shipping_street,
        city=row.shipping_city,
        postal_code=row.shipping_postal,
        country_code=row.shipping_country,
    )
    order = Order(
        order_id=OrderId(row.order_id),
        customer_id=row.customer_id,
        shipping_address=address,
        currency=row.currency,
    )
    # Restore internal state via object.__setattr__ (bypasses frozen/init)
    object.__setattr__(order, "status", row.status)
    object.__setattr__(order, "_version", row.version)
    lines_data = json.loads(row.lines_json)
    restored_lines = [
        OrderLine(
            product_id=ProductId(item["product_id"]),
            product_name=item["product_name"],
            quantity=item["quantity"],
            unit_price=Money(Decimal(item["unit_price"]), row.currency),
        )
        for item in lines_data
    ]
    object.__setattr__(order, "_lines", restored_lines)
    return order


def _to_orm(order: Order) -> dict:
    lines_data = [
        {
            "product_id":   line.product_id.value,
            "product_name": line.product_name,
            "quantity":     line.quantity,
            "unit_price":   str(line.unit_price.amount),
        }
        for line in order.lines
    ]
    return {
        "order_id":         str(order.order_id),
        "customer_id":      order.customer_id,
        "status":           order.status,
        "currency":         order.currency,
        "shipping_street":  order.shipping_address.street,
        "shipping_city":    order.shipping_address.city,
        "shipping_postal":  order.shipping_address.postal_code,
        "shipping_country": order.shipping_address.country_code,
        "lines_json":       json.dumps(lines_data),
        "version":          order._version,
    }


class SQLAlchemyOrderRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def get(self, order_id: str) -> Optional[Order]:
        row = self._session.get(OrderORM, order_id)
        return _to_domain(row) if row else None

    def save(self, order: Order) -> None:
        data = _to_orm(order)
        existing = self._session.get(OrderORM, data["order_id"])
        if existing:
            for k, v in data.items():
                setattr(existing, k, v)
        else:
            self._session.add(OrderORM(**data))

    def next_id(self) -> str:
        return str(uuid.uuid4())

    def find_by_customer(self, customer_id: str) -> list[Order]:
        stmt = select(OrderORM).where(OrderORM.customer_id == customer_id)
        rows = self._session.execute(stmt).scalars().all()
        return [_to_domain(row) for row in rows]

In-Memory Repository (for Tests and Local Dev)

# tests/fakes.py
from __future__ import annotations
from typing import Optional
import uuid
import copy
from domain.order_aggregate import Order


class InMemoryOrderRepository:
    """
    No-I/O repository. Runs in microseconds.
    Used in all unit tests and optionally in local development.
    """

    def __init__(self) -> None:
        self._store: dict[str, Order] = {}

    def get(self, order_id: str) -> Optional[Order]:
        order = self._store.get(order_id)
        return copy.deepcopy(order) if order else None  # simulate isolation

    def save(self, order: Order) -> None:
        self._store[str(order.order_id)] = copy.deepcopy(order)

    def next_id(self) -> str:
        return str(uuid.uuid4())

    def find_by_customer(self, customer_id: str) -> list[Order]:
        return [o for o in self._store.values() if o.customer_id == customer_id]

    # Test helpers
    def all(self) -> list[Order]:
        return list(self._store.values())

    def count(self) -> int:
        return len(self._store)

Part 6 - Domain Events

A Domain Event is something that happened in the domain that the domain experts care about.

Events are facts. OrderPlaced happened. The past is immutable, so events are immutable (frozen=True). Events decouple the aggregate from its side effects - the aggregate does not know who is listening.

Defining Events

# domain/events.py
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
import uuid


def _now() -> datetime:
    return datetime.now(timezone.utc)

def _event_id() -> str:
    return str(uuid.uuid4())


@dataclass(frozen=True)
class DomainEvent:
    """All domain events are immutable records of something that happened."""
    event_id: str = field(default_factory=_event_id, compare=False)
    occurred_at: datetime = field(default_factory=_now, compare=False)


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str = ""
    customer_id: str = ""
    total_amount: str = ""
    currency: str = ""


@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
    order_id: str = ""


@dataclass(frozen=True)
class OrderShipped(DomainEvent):
    order_id: str = ""
    tracking_number: str = ""


@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
    order_id: str = ""
    reason: str = ""
    previous_status: str = ""


@dataclass(frozen=True)
class PaymentReceived(DomainEvent):
    order_id: str = ""
    amount: str = ""
    currency: str = ""
    payment_reference: str = ""

In-Process Event Bus

# domain/event_bus.py
from __future__ import annotations
from typing import Callable, Type, TypeVar
from collections import defaultdict
from domain.events import DomainEvent

E = TypeVar("E", bound=DomainEvent)


class EventBus:
    """
    Synchronous in-process event bus.

    For distributed eventing (Kafka, RabbitMQ, SNS) replace only the
    publish implementation - subscribers remain unchanged.
    """

    def __init__(self) -> None:
        self._handlers: dict[type, list[Callable]] = defaultdict(list)

    def subscribe(self, event_type: Type[E], handler: Callable[[E], None]) -> None:
        self._handlers[event_type].append(handler)  # type: ignore[arg-type]

    def publish(self, event: DomainEvent) -> None:
        for handler in self._handlers.get(type(event), []):
            handler(event)

    def publish_all(self, events: list[DomainEvent]) -> None:
        for event in events:
            self.publish(event)

Wiring Events to Side Effects

# application/event_handlers.py
from domain.events import OrderPlaced, OrderShipped, OrderCancelled
from domain.event_bus import EventBus


def on_order_placed(event: OrderPlaced) -> None:
    # Send confirmation email
    print(f"[EMAIL]     Order {event.order_id} placed - sending confirmation")
    # Notify warehouse
    print(f"[WAREHOUSE] Order {event.order_id} added to fulfilment queue")
    # Update analytics
    print(f"[ANALYTICS] Revenue event: {event.currency} {event.total_amount}")


def on_order_shipped(event: OrderShipped) -> None:
    print(f"[EMAIL]     Order {event.order_id} shipped - tracking: {event.tracking_number}")
    print(f"[COURIER]   Activate tracking for {event.tracking_number}")


def on_order_cancelled(event: OrderCancelled) -> None:
    print(f"[REFUND]    Processing refund for order {event.order_id}")
    print(f"[EMAIL]     Cancellation notice - reason: {event.reason}")
    print(f"[INVENTORY] Releasing stock for cancelled order {event.order_id}")


def register_handlers(bus: EventBus) -> None:
    bus.subscribe(OrderPlaced,    on_order_placed)
    bus.subscribe(OrderShipped,   on_order_shipped)
    bus.subscribe(OrderCancelled, on_order_cancelled)

The Order aggregate records OrderPlaced without knowing who handles it. Handlers are registered at startup in the application layer. This means you can add a new side effect (notify a third-party fulfilment system, update a dashboard) by adding one bus.subscribe call - the domain is not touched.

Part 7 - Bounded Contexts

A Bounded Context is an explicit boundary within which a particular domain model applies. Different models of the same concept coexist legitimately in different contexts.

The word "Order" means something different to every team. DDD says: accept this, make the boundaries explicit, and integrate via events.

# contexts/order_management/domain.py
# OrderManagement context: cares about customer intent, pricing, items.

from dataclasses import dataclass, field
from domain.value_objects import Money, Address, OrderLine


@dataclass(eq=False)
class Order:
    order_id: str
    customer_id: str
    shipping_address: Address
    lines: list[OrderLine]
    discount_code: str | None
    payment_method: str
    status: str   # pending, placed, confirmed, shipped, cancelled

# contexts/shipping/domain.py
# Shipping context: cares about physical movement, carrier, weight.
# Does NOT know about prices, discount codes, or payment methods.

from dataclasses import dataclass


@dataclass
class PhysicalItem:
    product_name: str
    weight_kg: float
    dimensions_cm: tuple[float, float, float]


@dataclass(eq=False)
class ShipmentOrder:
    shipment_id: str
    source_order_id: str    # reference into OrderManagement context
    recipient_name: str
    delivery_address: "Address"
    items: list[PhysicalItem]
    carrier: str
    tracking_number: str | None
    status: str   # pending_collection, in_transit, delivered, failed

# contexts/billing/domain.py
# Billing context: cares about amounts, invoicing, payment status.
# Does NOT know about physical dimensions or carrier names.

from dataclasses import dataclass
from datetime import datetime
from domain.value_objects import Money


@dataclass
class InvoiceLine:
    description: str
    quantity: int
    unit_price: Money
    line_total: Money


@dataclass(eq=False)
class Invoice:
    invoice_id: str
    source_order_id: str
    customer_id: str
    lines: list[InvoiceLine]
    total: Money
    payment_status: str   # pending, paid, refunded, disputed
    issued_at: datetime

Context Map (ASCII)

                    ┌──────────────────────────────────────┐
                    │         E-Commerce Platform          │
                    │                                      │
  HTTP ────────────▶│  ┌──────────────────────────────┐   │
  (Customer UI)     │  │  OrderManagement Context     │   │
                    │  │                              │   │
                    │  │  Order (full model)          │   │
                    │  │  Cart → Order lifecycle      │   │
                    │  │  Pricing, discount codes     │   │
                    │  └──────────┬───────────────────┘   │
                    │             │ OrderPlaced event      │
                    │      ┌──────┴────────────────┐      │
                    │      │  Anti-Corruption Layer │      │
                    │      │  (event translation)   │      │
                    │      └──────┬──────────┬──────┘      │
                    │             │          │              │
                    │    ┌────────▼──┐  ┌────▼──────────┐  │
                    │    │ Shipping  │  │   Billing     │  │
                    │    │ Context   │  │   Context     │  │
                    │    │           │  │               │  │
                    │    │ Shipment  │  │   Invoice     │  │
                    │    │ Carrier   │  │   Payment     │  │
                    │    │ Tracking  │  │   Refund      │  │
                    │    └───────────┘  └───────────────┘  │
                    └──────────────────────────────────────┘

Context Integration

# integration/context_bridges.py
from domain.events import OrderPlaced
from domain.event_bus import EventBus


def register_cross_context_handlers(bus: EventBus) -> None:
    bus.subscribe(OrderPlaced, _create_shipping_order)
    bus.subscribe(OrderPlaced, _create_billing_invoice)


def _create_shipping_order(event: OrderPlaced) -> None:
    """
    Translates an OrderManagement event into a Shipping context action.
    The Shipping context builds its own ShipmentOrder - it does not share
    the OrderManagement Order object. This is the Anti-Corruption Layer.
    """
    # Fetch details from OrderManagement read model if needed
    print(f"[Shipping] Creating ShipmentOrder for source order {event.order_id}")
    # ... build ShipmentOrder from event data ...


def _create_billing_invoice(event: OrderPlaced) -> None:
    """Translates to a Billing context Invoice."""
    print(f"[Billing] Creating Invoice for order {event.order_id}, "
          f"amount {event.currency} {event.total_amount}")

The ShipmentOrder and the Invoice are not the same object as the OrderManagement Order. Each context owns its own model, optimised for its own needs. Integration happens through events, never through shared database tables or shared domain objects.

Part 8 - Application Services

Application Services are the thin orchestration layer between the outside world and the domain. They do not contain business logic - that lives in aggregates. They coordinate loading, calling, saving, and publishing.

# application/use_cases.py
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from domain.order_aggregate import Order
from domain.entities import OrderId
from domain.value_objects import Address
from domain.repositories import OrderRepository
from domain.event_bus import EventBus


# ── Command DTOs ──────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class PlaceOrderCommand:
    customer_id: str
    shipping_street: str
    shipping_city: str
    shipping_postal_code: str
    shipping_country_code: str
    currency: str
    lines: list   # [{"product_id", "product_name", "quantity", "unit_price"}]


@dataclass(frozen=True)
class PlaceOrderResult:
    order_id: str
    total_amount: str
    currency: str
    status: str


# ── Use Cases ─────────────────────────────────────────────────────────────────

class PlaceOrderUseCase:
    """
    Orchestrates: create Order aggregate → add lines → place → save → publish events.
    Contains zero business rules. All invariants enforced inside Order.
    """

    def __init__(self, repo: OrderRepository, bus: EventBus) -> None:
        self._repo = repo
        self._bus = bus

    def execute(self, cmd: PlaceOrderCommand) -> PlaceOrderResult:
        order_id = OrderId(self._repo.next_id())
        address = Address(
            street=cmd.shipping_street,
            city=cmd.shipping_city,
            postal_code=cmd.shipping_postal_code,
            country_code=cmd.shipping_country_code,
        )
        order = Order(
            order_id=order_id,
            customer_id=cmd.customer_id,
            shipping_address=address,
            currency=cmd.currency,
        )

        for item in cmd.lines:
            order.add_line(
                product_id=item["product_id"],
                product_name=item["product_name"],
                quantity=item["quantity"],
                unit_price=Decimal(str(item["unit_price"])),
            )

        order.place()              # domain enforces "must have lines"
        self._repo.save(order)
        self._bus.publish_all(order.pop_events())

        return PlaceOrderResult(
            order_id=str(order.order_id),
            total_amount=str(order.total.amount),
            currency=order.currency,
            status=order.status,
        )


class ShipOrderUseCase:
    def __init__(self, repo: OrderRepository, bus: EventBus) -> None:
        self._repo = repo
        self._bus = bus

    def execute(self, order_id: str, tracking_number: str) -> None:
        order = self._repo.get(order_id)
        if order is None:
            raise ValueError(f"Order {order_id!r} not found.")
        order.confirm()              # domain enforces valid transition
        order.ship(tracking_number)  # domain enforces tracking number required
        self._repo.save(order)
        self._bus.publish_all(order.pop_events())


class CancelOrderUseCase:
    def __init__(self, repo: OrderRepository, bus: EventBus) -> None:
        self._repo = repo
        self._bus = bus

    def execute(self, order_id: str, reason: str) -> None:
        order = self._repo.get(order_id)
        if order is None:
            raise ValueError(f"Order {order_id!r} not found.")
        order.cancel(reason)         # domain enforces cannot cancel shipped
        self._repo.save(order)
        self._bus.publish_all(order.pop_events())

Complete Test Suite - Zero I/O

# tests/test_order_use_cases.py
import pytest
from decimal import Decimal
from application.use_cases import (
    PlaceOrderUseCase, PlaceOrderCommand, PlaceOrderResult,
    ShipOrderUseCase, CancelOrderUseCase,
)
from domain.events import OrderPlaced, OrderShipped, OrderCancelled
from domain.event_bus import EventBus
from tests.fakes import InMemoryOrderRepository


def _make_cmd(**overrides) -> PlaceOrderCommand:
    defaults = dict(
        customer_id="cust-123",
        shipping_street="1 Main St",
        shipping_city="London",
        shipping_postal_code="EC1A 1AA",
        shipping_country_code="GB",
        currency="GBP",
        lines=[
            {"product_id": "P1", "product_name": "Python Course", "quantity": 1, "unit_price": "49.00"},
            {"product_id": "P2", "product_name": "AI Course",     "quantity": 2, "unit_price": "29.00"},
        ],
    )
    defaults.update(overrides)
    return PlaceOrderCommand(**defaults)


def _setup():
    repo = InMemoryOrderRepository()
    bus  = EventBus()
    return repo, bus


def test_place_order_computes_total():
    repo, bus = _setup()
    result = PlaceOrderUseCase(repo, bus).execute(_make_cmd())
    assert result.status == "placed"
    assert result.total_amount == "107.00"   # 49 + 2×29


def test_place_order_persists():
    repo, bus = _setup()
    result = PlaceOrderUseCase(repo, bus).execute(_make_cmd())
    saved = repo.get(result.order_id)
    assert saved is not None
    assert saved.line_count == 2


def test_place_order_publishes_event():
    repo, bus = _setup()
    received: list[OrderPlaced] = []
    bus.subscribe(OrderPlaced, received.append)
    result = PlaceOrderUseCase(repo, bus).execute(_make_cmd())
    assert len(received) == 1
    assert received[0].order_id == result.order_id
    assert received[0].currency == "GBP"


def test_cannot_place_empty_order():
    repo, bus = _setup()
    with pytest.raises(ValueError, match="no items"):
        PlaceOrderUseCase(repo, bus).execute(_make_cmd(lines=[]))


def test_ship_order_full_lifecycle():
    repo, bus = _setup()
    shipped: list[OrderShipped] = []
    bus.subscribe(OrderShipped, shipped.append)

    place_uc = PlaceOrderUseCase(repo, bus)
    result   = place_uc.execute(_make_cmd())

    ShipOrderUseCase(repo, bus).execute(result.order_id, "TRACK-GB-9999")

    order = repo.get(result.order_id)
    assert order.status == "shipped"
    assert len(shipped) == 1
    assert shipped[0].tracking_number == "TRACK-GB-9999"


def test_cancel_placed_order():
    repo, bus = _setup()
    cancelled: list[OrderCancelled] = []
    bus.subscribe(OrderCancelled, cancelled.append)

    result = PlaceOrderUseCase(repo, bus).execute(_make_cmd())
    CancelOrderUseCase(repo, bus).execute(result.order_id, "Customer changed mind")

    order = repo.get(result.order_id)
    assert order.status == "cancelled"
    assert cancelled[0].reason == "Customer changed mind"
    assert cancelled[0].previous_status == "placed"


def test_cannot_cancel_shipped_order():
    repo, bus = _setup()
    result = PlaceOrderUseCase(repo, bus).execute(_make_cmd())
    ShipOrderUseCase(repo, bus).execute(result.order_id, "TRACK-001")

    with pytest.raises(ValueError):
        CancelOrderUseCase(repo, bus).execute(result.order_id, "Too late")

DDD Building Blocks - Reference Table

Building Block	Python Construct	Equality	Mutable	Key Rule
Entity	`@dataclass(eq=False)` + custom `__eq__` on ID	By ID	Yes	Never expose internal collections directly
Value Object	`@dataclass(frozen=True)`	By value	Never	Validate all inputs in `__post_init__`
Aggregate Root	`@dataclass(eq=False)`	By ID	Yes	All mutations via public methods; emit events
Repository	`Protocol` or `ABC` in domain; class in infra	-	-	Domain imports Protocol only; no ORM in domain
Domain Event	`@dataclass(frozen=True)`	By value	Never	Emitted by aggregates; handled by application layer
Application Service	Plain class	-	-	Orchestrates only; zero business logic
Bounded Context	Python package	-	-	One model per context; integrate via events

Interview Patterns

Pattern 1 - "What is the difference between an Entity and a Value Object?"

Strong answer: "An Entity has identity that persists through state changes - Order #4521 is the same order whether its status is pending or shipped. A Value Object has no identity beyond its value - two Money(50, 'USD') instances are interchangeable and equal. In Python, Entities use @dataclass(eq=False) with a custom __eq__ comparing IDs; Value Objects use @dataclass(frozen=True) which gives immutability and auto-generates __hash__ from all fields."

Pattern 2 - "Why should the domain layer not import SQLAlchemy?"

Strong answer: "Business rules evolve on product timelines - quarterly. Infrastructure evolves on technical timelines - when the team decides to change database. If the domain imports SQLAlchemy, migrating to a different ORM or switching to an event store requires understanding and editing business logic. By keeping the domain pure - only standard library and your own domain types - you can run every business rule test without a database, and you can swap storage technology without touching a single domain class."

Pattern 3 - "What is a Bounded Context?"

Strong answer: "A Bounded Context is an explicit boundary within which one domain model is consistent. The word 'Order' means different things to a warehouse team (items to pick), a billing team (amount to charge), and a logistics team (package to route). DDD says to accept this ambiguity and make it explicit: each context has its own model, optimised for its own needs. The contexts integrate via domain events, not via shared objects or shared database tables. The boundary prevents one team's model complexity from leaking into another team's codebase."

Pattern 4 - "Where should business rules live?"

Strong answer: "In the domain aggregate. The rule 'you cannot cancel a shipped order' lives in Order.cancel(), not in a service class. This means it is enforced at every call site automatically - you cannot bypass it by calling order.status = 'cancelled' because status is not a public field. The application service's job is to coordinate: load from repo, call domain method, save, publish events. It contains zero business logic itself."

Pattern 5 - "Walk me through placing an order end-to-end"

Strong answer covers all six steps:

HTTP request arrives, controller translates to PlaceOrderCommand DTO
PlaceOrderUseCase.execute(cmd) is called (application layer)
Use case creates an Order aggregate, calls add_line() for each item
Use case calls order.place() - aggregate enforces "must have lines", records OrderPlaced event
Use case calls repo.save(order) and bus.publish_all(order.pop_events())
Event handlers send confirmation email, notify warehouse, update analytics - none of this is in the aggregate
Use case returns PlaceOrderResult DTO; controller returns HTTP 201

What You Will Learn​

Prerequisites​

Part 1 - Ubiquitous Language​

The Communication Breakdown​

Building the Glossary​

What Bad Naming Costs​

Part 2 - Entities​

Python Implementation​

Why eq=False Is the Key Decision​

Part 3 - Value Objects​

Python Implementation​

Value Object vs Entity Decision Guide​

Part 4 - Aggregates​

The Order Aggregate​

Invariant Enforcement Demo​

Part 5 - Repositories​

Abstract Repository (Domain Layer)​

Infrastructure - SQLAlchemy Repository​

In-Memory Repository (for Tests and Local Dev)​

Part 6 - Domain Events​

Defining Events​

In-Process Event Bus​

Wiring Events to Side Effects​

Part 7 - Bounded Contexts​

Context Map (ASCII)​

Context Integration​

Part 8 - Application Services​

Complete Test Suite - Zero I/O​

DDD Building Blocks - Reference Table​

Interview Patterns​

Pattern 1 - "What is the difference between an Entity and a Value Object?"​

Pattern 2 - "Why should the domain layer not import SQLAlchemy?"​

Pattern 3 - "What is a Bounded Context?"​

Pattern 4 - "Where should business rules live?"​

Pattern 5 - "Walk me through placing an order end-to-end"​

What You Will Learn

Prerequisites

Part 1 - Ubiquitous Language

The Communication Breakdown

Building the Glossary

What Bad Naming Costs

Part 2 - Entities

Python Implementation

Why `eq=False` Is the Key Decision

Part 3 - Value Objects

Python Implementation

Value Object vs Entity Decision Guide

Part 4 - Aggregates

The `Order` Aggregate

Invariant Enforcement Demo

Part 5 - Repositories

Abstract Repository (Domain Layer)

Infrastructure - SQLAlchemy Repository

In-Memory Repository (for Tests and Local Dev)

Part 6 - Domain Events

Defining Events

In-Process Event Bus

Wiring Events to Side Effects

Part 7 - Bounded Contexts

Context Map (ASCII)

Context Integration

Part 8 - Application Services

Complete Test Suite - Zero I/O

DDD Building Blocks - Reference Table

Interview Patterns

Pattern 1 - "What is the difference between an Entity and a Value Object?"

Pattern 2 - "Why should the domain layer not import SQLAlchemy?"

Pattern 3 - "What is a Bounded Context?"

Pattern 4 - "Where should business rules live?"

Pattern 5 - "Walk me through placing an order end-to-end"